package spring.cloud.tasks.client.item_task_processor.pooled.impl;

import com.google.common.collect.Interner;
import com.google.common.collect.Interners;
import spring.cloud.tasks.client.item_task_processor.pooled.TaskPauseOrResumeScheduleTaskPooled;
import spring.cloud.tasks.client.item_task_processor.pooled.timer_task.PauseOrResumeScheduleTask;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;


public class TaskPauseOrResumeScheduleTaskPooledImpl implements TaskPauseOrResumeScheduleTaskPooled {
    public static final String resume = "resume";
    public static final String pause = "pause";

    private static ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(50);

    private static Map<String, ScheduledFuture> resumeScheduleTaskIdToScheduledFutureMap = new HashMap<String, ScheduledFuture>();
    private static Map<String, ScheduledFuture> pauseScheduleTaskIdToScheduledFutureMap = new HashMap<String, ScheduledFuture>();
    private Interner<String> resumeInterner = Interners.newWeakInterner();
    private Interner<String> pauseInterner = Interners.newWeakInterner();

    @Override
    public void init(String type, PauseOrResumeScheduleTask pauseOrResumeScheduleTask, Date nextTime) {
        if (resume.equals(type)) {
            resumeScheduleTaskIdToScheduledFutureMap.put(pauseOrResumeScheduleTask.getProcessorId(), null);
            ScheduledFuture scheduledFuture = this.schedule(type, pauseOrResumeScheduleTask, nextTime);
            resumeScheduleTaskIdToScheduledFutureMap.put(pauseOrResumeScheduleTask.getProcessorId(), scheduledFuture);
        } else if (pause.equals(type)) {

        }

    }

    @Override
    public ScheduledFuture schedule(String type, PauseOrResumeScheduleTask pauseOrResumeScheduleTask, Date nextTime) {
        if (resume.equals(type)) {
            synchronized (resumeInterner.intern(pauseOrResumeScheduleTask.getProcessorId())) {
                if (resumeScheduleTaskIdToScheduledFutureMap.containsKey(pauseOrResumeScheduleTask.getProcessorId())) {
                    return scheduledExecutorService.schedule(pauseOrResumeScheduleTask, (nextTime.getTime() - new Date().getTime()), TimeUnit.MILLISECONDS);
                }
            }
        } else if (pause.equals(type)) {
            synchronized (pauseInterner.intern(pauseOrResumeScheduleTask.getProcessorId())) {
                if (pauseScheduleTaskIdToScheduledFutureMap.containsKey(pauseOrResumeScheduleTask.getProcessorId())) {
                    return scheduledExecutorService.schedule(pauseOrResumeScheduleTask, (nextTime.getTime() - new Date().getTime()), TimeUnit.MILLISECONDS);
                }
            }
        }

        return null;
    }

    @Override
    public void cancel(String type, PauseOrResumeScheduleTask pauseOrResumeScheduleTask) {
        if (resume.equals(type)) {
            synchronized (resumeInterner.intern(pauseOrResumeScheduleTask.getProcessorId())) {
                if (resumeScheduleTaskIdToScheduledFutureMap.containsKey(pauseOrResumeScheduleTask.getProcessorId())) {
                    ScheduledFuture scheduledFuture = resumeScheduleTaskIdToScheduledFutureMap.remove(pauseOrResumeScheduleTask.getProcessorId());
                    if (scheduledFuture != null) {
                        scheduledFuture.cancel(true);
                    }
                }
            }
        } else if (pause.equals(type)) {
            synchronized (pauseInterner.intern(pauseOrResumeScheduleTask.getProcessorId())) {
                if (pauseScheduleTaskIdToScheduledFutureMap.containsKey(pauseOrResumeScheduleTask.getProcessorId())) {
                    ScheduledFuture scheduledFuture = pauseScheduleTaskIdToScheduledFutureMap.remove(pauseOrResumeScheduleTask.getProcessorId());
                    if (scheduledFuture != null) {
                        scheduledFuture.cancel(true);
                    }
                }
            }
        }

    }
}
